package com.atguigu.gmall.realtime.app.marketing.func;

import com.atguigu.gmall.realtime.app.marketing.bean.GmallEvent;
import com.atguigu.gmall.realtime.app.marketing.bean.MarketingMessage;
import com.atguigu.gmall.realtime.app.marketing.bean.MarketingRuleAllInfo;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.RedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

public class MarketingRuleProcessor extends ProcessFunction<GmallEvent, MarketingMessage> {

    Timer timer;

    RuleFetcher ruleFetcher;

    RuleMatcher ruleMatcher;

    MessageBuilder messageBuilder;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.MYSQL_DRIVER);
        Connection connection = DriverManager.getConnection(GmallConfig.MYSQL_SERVER);

        Jedis jedis = RedisUtil.getJedis();

        ruleFetcher=new RuleFetcher(connection,jedis);
        ruleMatcher=new RuleMatcher(jedis);
        messageBuilder=new MessageBuilder();
        timer=new Timer();
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                List<MarketingRuleAllInfo> marketingRuleAllInfoList = ruleFetcher.fetch();
                ruleMatcher.setRuleAllInfoList(marketingRuleAllInfoList);
            }
        };
        //time 调度timetask
        timer.schedule(timerTask,0,5000);

    }

    @Override
    public void processElement(GmallEvent event, ProcessFunction<GmallEvent, MarketingMessage>.Context ctx, Collector<MarketingMessage> out) throws Exception {
        //规则匹配操作
        List<MarketingRuleAllInfo> matchedRuleList = ruleMatcher.match(event);
        //消息构造
        if(matchedRuleList.size()>0){
            for (MarketingRuleAllInfo marketingRuleAllInfo : matchedRuleList) {
                MarketingMessage message = messageBuilder.build(marketingRuleAllInfo, event);
                out.collect(message);
            }
        }


    }
}
